package Autodesk;


import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ProccessDynArchiv extends Configured implements Tool  {
	
		public Path prodPath;
		public static class LogMapper extends Mapper<Object, Text, Text, IntWritable> {
			
			private final static IntWritable one = new IntWritable(1);
			private Text word = new Text();
			private Map<ArrayList<String>,String> staticList=new HashMap<ArrayList<String>,String>();
			private String param;
			boolean AreProducts=true;
			public void setup(Context context){
				Configuration conf=context.getConfiguration();
				 if (conf.getBoolean("arch.patterns", false)) {
					  Path[] patternsFiles = new Path[0];
					  try{
				           patternsFiles = DistributedCache.getLocalCacheFiles(conf);
				           String paramNotReady=conf.get("input.parameter");
				           param=paramNotReady.toString().toLowerCase();
					  }catch(IOException ioe){
				           System.err.println("Caught exception while getting cached files: ");
					  }
					  for (Path patternsFile : patternsFiles) {
						  	getStaticList(patternsFile,staticList,param); 
					  }
					 
				 }
			}
			//Parse the cached file
			 //We consider that the input mapping product document is: ID-ProdAb/1/2/3/4/5 - ReleaseAb
			 //We consider that the input mapping channel/language document is: ID-ProdAb/1/2/3/4/5
			private void getStaticList(Path patternFile,Map<ArrayList<String>,String> staticList,String param){
				try{
					BufferedReader fis=new BufferedReader(new FileReader(patternFile.toString()));	
					String pattern=null;
					
					while((pattern =fis.readLine())!=null){
						String record[] = pattern.split("\t");
						if("products".equals(param) || "product".equals(param)){ 
							int i=1;
							ArrayList<String> abbProduct=new ArrayList<String>();
							while(i<record.length-1){
								if(!record[i].equals("")){
									abbProduct.add(record[i]);
								}
								i++;
							}
							abbProduct.add(record[record.length-1]); //The end of the list is the release abbreviation
							staticList.put(abbProduct,record[0]);
						}else{
							int i=1;
							ArrayList<String> abbMatcher=new ArrayList<String>();
							while(record.length>i){
							//	abbMatcher.add(record[i].toLowerCase());
								abbMatcher.add(record[i]);
								i++;
							}
							staticList.put(abbMatcher,record[0]);
						}
					}
					fis.close();
				}catch(IOException ioe){
			         System.err.println("Caught exception while parsing the cached file '");
				}
			}
			
			private String lookForKey(Map<ArrayList<String>,String> staticList,String logkey){
				boolean found=false;
				String id="";
				Iterator it = staticList.keySet().iterator();
				while(it.hasNext() & !found) {	
					ArrayList<String> keyList = (ArrayList<String>) it.next();
					for(int i=0;i<keyList.size() & !found;i++){
						if(logkey.toLowerCase().equals((String)keyList.get(i).toLowerCase())){
							id=staticList.get(keyList);
							found=true;
						}
					}
				}
				return id;
			}
			private static String getYear(String value){
				String year="";
				Pattern yearPattern = Pattern
						.compile("20[0-9][0-9]");
				Matcher yearMatcher = yearPattern.matcher(value);
				year = yearMatcher.find() ? yearMatcher.group() : value;
				return year;
			}
			private String lookForKeyProduct(Map<ArrayList<String>,String> staticList,String product,String release){
				boolean found=false;
				String id="";
				Iterator it = staticList.keySet().iterator();
				int g=0;
				while(it.hasNext() & !found) {	
					ArrayList<String> keyList = (ArrayList<String>) it.next();
					for(int i=0;i<keyList.size() & !found;i++){
					//	System.out.println(g+"-Compare:"+product.toLowerCase()+"  with "+keyList.get(i).toLowerCase());
						if(product.toLowerCase().equals((String)keyList.get(i).toLowerCase())){
							for(int f=0;f<keyList.size() & !found;f++){
							//	System.out.println(g+"-CompareRELEASE:"+release+"  with "+keyList.get(f));
								if(release.equals("all")){
									if(release.equals(keyList.get(f))){
										found=true;
										id=staticList.get(keyList);
									}
								}else{
									if(getYear(release).equals( getYear(keyList.get(f))) ){
										found=true;
										id=staticList.get(keyList);
									}
								}
							}
						}
					}
					g++;
				}
		//		System.out.println(id);
				return id;
			}

			/**
			 * @param line
			 * @param year
			 * @param month
			 * @param day
			 */
			private void processDate(String line, String year, String month,
					String day) {
				String date="";
				String shortdate;
				String dateMapper;
				Pattern datePattern = Pattern
						.compile("[0-3][0-9][/][a-zA-Z]*[/][0-9]{4}:[0-9]{2}:[0-9]{2}:[0-9]{2}.[0-9]{3}");
				Matcher dateMatcher = datePattern.matcher(line);
				while (dateMatcher.find()) {
					date = dateMatcher.group();
					if (date != null) {
						shortdate = date.substring(0, date.indexOf(":"));

						StringTokenizer datetokenizer = new StringTokenizer(
								shortdate, "/");
						while (datetokenizer.hasMoreTokens()) {
							String token = datetokenizer.nextToken();
							if (day == null)
								day = token;
							else if (month == null)
								month = token;
							else if (year == null)
								year = token;
							else
								break;
						}
					}
				}
				dateMapper=year+"-"+month+"-"+day;
				if(("time".equals(param)) || ("interval".equals(param))){
					if(staticList.containsKey(dateMapper)){
						date=staticList.get(dateMapper);
					}else{
						word.set(date);
					}
				}
			}
			/**
			 * @param line
			 * @return
			 */
			private String processChannels(String line) {
				String channel;
				Pattern channelPattern = Pattern
						.compile("project=[\\w\'-]*|[/]resource[/][a-zA-Z0-9]*[/]");
				Matcher channelMatcher = channelPattern.matcher(line);
				channel = channelMatcher.find() ? channelMatcher.group() : "";
				if (!channel.equals("") && channel.indexOf("project=") > -1)
					channel = channel.substring(channel.indexOf("=") + 1);
				else if (!channel.equals("") && channel.indexOf("/resource/") > -1)
					channel = channel
							.substring(
									channel.indexOf("/",
											channel.indexOf("/resource/") + 1) + 1,
									channel.lastIndexOf("/"));
				if(("channel".equals(param)) || ("channels".equals(param)) ){
					if(lookForKey(staticList, channel).equals("")){
						word.set(channel);
					}
				}
				return channel;
			}
			/**
			 * @param token
			 */
			private void processLanguages(String queryLine,String token) {
				String tokenSub;
				tokenSub=token.substring(2);
				String isSpace=tokenSub.substring(0,1);
				if(lookForKey(staticList, tokenSub).equals("") && !(isSpace.equals("\t"))){
					//tokenSub=generateKey(queryLine,token, tokenSub);
					word.set(tokenSub);
				}
			}
			/**
			 * @param queryLine
			 * @param token
			 * @param tokenProd
			 * @throws UnsupportedEncodingException
			 */
			private void proccesProduct(String queryLine, String token,
					String tokenProd) throws UnsupportedEncodingException {
				String tokenSub;
				if(!(tokenProd.equals("null")) && (tokenProd!=null) ){ //IT EXCLUDES when p=null
					if((tokenProd.indexOf("<")>-1 )){
						while(tokenProd.indexOf("<")>-1){
						tokenProd=tokenProd.substring(tokenProd.indexOf(">")+1,tokenProd.lastIndexOf("</"));
						}
					}
					
					StringTokenizer tokenizer2 = new StringTokenizer(queryLine, "&");
					while ((tokenizer2.hasMoreTokens()) & (token.indexOf("v=") < 0 )){//We have to code this because the version token could have been written before the product token.
						token=tokenizer2.nextToken();
					}
					String tokenRele="";
					if ( (token.indexOf("v=") > -1 )){
						tokenRele=token.substring(2);	
					}
					tokenRele=URLDecoder.decode(tokenRele,"UTF-8");
					//System.out.println(tokenRele);
					if(tokenRele.equals("")){
						tokenRele="all";
					}
					if(tokenRele.indexOf("<")>-1 ){
						while(tokenRele.indexOf("<")>-1){
						tokenRele=tokenRele.substring(tokenRele.indexOf(">")+1,tokenRele.lastIndexOf("</"));
						}
					}
					
					if((lookForKeyProduct(staticList, tokenProd,tokenRele).equals("")) ){
						tokenSub = generateKeyProducts(queryLine,tokenProd,tokenRele);
						word.set(tokenSub); //We find a channel unsaved in the database
					}
				}
			}
			/**
			 * If "origin=upi" is written in the log files it means that the product string is an abbreviation.If it is not written, it is the baseline value
			 * @param tokenProd
			 * 
			 * @param tokenRele
			 * @return
			 */
			private String generateKeyProducts(String queryline,String tokenProd, String tokenRele) {
				String tokenSub;
				if(queryline.indexOf("origin=upi")<0 ){//IT IS A BASELINE VALUE
					tokenSub=tokenProd+"\t"+""+"\t"+tokenRele;
				}else{
					tokenSub=""+"\t"+tokenProd+"\t"+tokenRele;
				}
				return tokenSub;
			}
			private String generateKey(String queryLine,String token,String tokenProd) {
				String tokenSub;
				if(queryLine.indexOf("origin=upi")<0 ){
					tokenSub=tokenProd+"\t"+"\t";
				}else{
					tokenSub="\t"+"\t"+tokenProd;
				}
				return tokenSub;
			}
			
			public void map(Object key, Text value, Context context)
					throws IOException, InterruptedException {
				
				//String line = value.toString().toLowerCase();
				String line = value.toString();

				// supportchannel2 -- fs.default.name
				
				Hashtable<String, String> ht = new Hashtable<String, String>();
				String channel = null;
				String year = null;
				String month = null;
				String day = null;
				
				//Date format[19/Aug/2012:20:29:49.626]
				processDate(line, year, month, day);
				// System.out.println ("date *--* " + date);
				//line = line.toLowerCase();
				String queryLine = "";
				if (line.indexOf("\"") > -1
						&& line.indexOf("\"", line.indexOf("\"") + 5) > -1)
					queryLine = line.substring(line.indexOf("\"") + 1,
							line.lastIndexOf("\""));
				if (queryLine.lastIndexOf(" ") > -1) {
					queryLine = queryLine.substring(0, queryLine.lastIndexOf(" "));
				}
				queryLine = queryLine.substring(queryLine.lastIndexOf("?") + 1);
/*				
				//Looking for IP.
				Pattern ipPattern = Pattern
						.compile("[0-9]{1,4}[.][0-9]{1,4}[.][0-9]{1,4}[.][0-9]{1,4}[:][0-9]{0,5}");
				Matcher ipMatcher = ipPattern.matcher(line);
				String 	ipPort = ipMatcher.find() ? ipMatcher.group() : "";
				String port = ipPort.substring(ipPort.indexOf(":")+1);
				String ip = ipPort.substring(0,ipPort.indexOf(":"));
*/				
				//Looking for CHANNEL.
				channel = processChannels(line);
				if (//(channel.indexOf("adsk-community") > -1) ||
						 (channel.indexOf("taxonomy")) > -1
						|| (channel.indexOf("productchannel") > -1)
						|| (channel.indexOf("productchannel2") > -1)
						|| (channel.indexOf("query-meta") > -1)
						|| (channel.indexOf("supportchannel") > -1)
						|| (channel.indexOf("supportchannel2") > -1)
						|| (channel.indexOf("acadconnectchannel") > -1)
						|| (channel.indexOf("acadconnectchannel2") > -1)) {
					// skip
					ht = null;
				} else 	{
					if ( queryLine.lastIndexOf("&") > -1) {
						String printstr = "";
						StringTokenizer tokenizer = new StringTokenizer(queryLine, "&");
						while (tokenizer.hasMoreTokens()) {
							String token = tokenizer.nextToken();
							printstr += token + "  |  ";
							if (token.indexOf("p=") > -1 || token.indexOf("v=") > -1
									|| token.indexOf("q=") > -1
									|| token.indexOf("l=") > -1
									|| token.indexOf("source=") > -1){
								if (token.indexOf("q=") > -1 && token.indexOf("*") > -1) {
									ht = null;
									break;
								}
								if (token.indexOf("q=") > -1) {
									ht.put("q", token.substring(2));
								} else if ( (token.indexOf("p=") > -1) & ( ("product".equals(param)) ||
										("products".equals(param)) ) ) {
									
									String tokenProd=token.substring(2);
									tokenProd=URLDecoder.decode(tokenProd, "UTF-8");
									proccesProduct(queryLine, token, tokenProd);
									
								} else if ( (token.indexOf("l=") > -1) & ( ("language".equals(param)) || ("languages".equals(param)) && (token.indexOf("l=http") < 0) ) ) {
									processLanguages(queryLine,token);
								} else if (token.indexOf("source=") > -1) {
									//ht.put("s", token.substring(7));

								}
							}
						}
						if (ht ==  null && printstr.indexOf("*") < 0) {
							System.out.println("channel -- " + channel + " queryLine: " +queryLine);
						}
					}

				}
				context.write(word, one);
			}
	}


	public static class LogReducer extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		
		private IntWritable result = new IntWritable();

		public void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			result.set(sum);
			context.write(key, result);
		
		}
	}
	
	public int run(String[] args) throws Exception { 
		Configuration conf = getConf();
		
		Job job = new Job(conf, "ParseLogs");
		job.setJarByClass(Autodesk.ProccessDynArchiv.class);
		
//		FileInputFormat.addInputPath(job, new Path(args[0]));
//		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		job.setMapperClass(LogMapper.class);
		job.setReducerClass(LogReducer.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		List<String> other_args = new ArrayList<String>();
		for (int i = 0; i < args.length; ++i) {
			if ("-arch".equals(args[i])) {
				prodPath=new Path(args[++i]);
				DistributedCache.addCacheFile(prodPath.toUri(),
						job.getConfiguration());
				job.getConfiguration().setBoolean("arch.patterns",
						true);
				job.getConfiguration().set("arch.path", prodPath.getName());
			} else if ("-param".equals(args[i])) {
				job.getConfiguration().set("input.parameter", new String(args[++i]));
			}else{
				other_args.add(args[i]);
			}
		}
	    FileInputFormat.setInputPaths(job, new Path(other_args.get(0)));
	    FileOutputFormat.setOutputPath(job, new Path(other_args.get(1)));

		System.exit(job.waitForCompletion(true)?0:1);
		
		return 0;
	
	}
	
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new ProccessDynArchiv(), args);
		System.exit(res);
		}

 
}